Node / ReticulumProjects / MeshChatX.git / files / tests / backend / test_conversation_summaries_landlock.py

Displaying Raw • Download

T8b949e# SPDX-License-Identifier: 0BSD

T8b949e"""Landlock coverage for conversation summary list queries."""

Tff7b72from T7ee787__future__ Tff7b72import Te6edf3annotations

Tff7b72import T7ee787subprocess

Tff7b72import T7ee787sys

Tff7b72import T7ee787textwrap

Tff7b72from T7ee787pathlib Tff7b72import Te6edf3Path

Tff7b72import T7ee787pytest

Tff7b72from T7ee787meshchatxT7ee787.T7ee787srcT7ee787.T7ee787backendT7ee787.T7ee787landlock_sandbox Tff7b72import Te6edf3landlock_kernel_supported

Tf0883e@pytestTff7b72.Td2a8ffmarkTff7b72.Td2a8ffskipifTb4b4b4(

Tff7b72not Te6edf3landlock_kernel_supportedTb4b4b4(Tb4b4b4)Tb4b4b4,

Te6edf3reasonTff7b72=Ta5d6ff"Ta5d6ffLandlock not available on this kernelTa5d6ff"Tb4b4b4,

Tb4b4b4)

Tff7b72def Td2a8fftest_landlock_conversation_summaries_list_query_okTb4b4b4(Tb4b4b4)Tb4b4b4:

T8b949e"""List-shaped summary queries must work under Landlock MEMORY temp."""

Te6edf3script Tff7b72= Te6edf3textwrapTff7b72.Td2a8ffdedentTb4b4b4(

Ta5d6ffrT8b949e"""

import os

import secrets

import sys

import tempfile

import time

from meshchatx.src.backend.database import Database

from meshchatx.src.backend.database.provider import DatabaseProvider

from meshchatx.src.backend.landlock_sandbox import apply_landlock_sandbox

from meshchatx.src.backend.message_handler import MessageHandler

td = tempfile.mkdtemp(prefix="ll_conv_sum_")

storage = os.path.join(td, "storage")

os.makedirs(storage)

os.environ["MESHCHAT_LANDLOCK"] = "1"

db_path = os.path.join(storage, "database.db")

if DatabaseProvider._instance is not None:

DatabaseProvider._instance.close_all()

DatabaseProvider._instance = None

db = Database(db_path)

db.initialize()

handler = MessageHandler(db)

for i in range(80):

peer = secrets.token_hex(16)

db.messages.upsert_lxmf_message(

{

"hash": secrets.token_hex(16),

"source_hash": peer,

"destination_hash": "localhashlocalhashlocalhashlo12",

"peer_hash": peer,

"state": "delivered",

"progress": 1.0,

"is_incoming": 1,

"method": "direct",

"delivery_attempts": 1,

"next_delivery_attempt_at": None,

"title": f"t{i}",

"content": "hello " + ("x" * 400),

"fields": "{}",

"timestamp": time.time() - i,

"rssi": -50,

"snr": 5.0,

"quality": 3,

"is_spam": 0,

"reply_to_hash": None,

}

)

ok = apply_landlock_sandbox(

storage_dir=storage,

reticulum_config_dir=storage,

log_dir=storage,

)

if not ok:

print("LANDLOCK_NOT_APPLIED")

sys.exit(2)

rows = handler.get_conversations("local", limit=50, offset=0)

if len(rows) != 50:

print("BAD_COUNT", len(rows))

sys.exit(3)

for row in rows:

row = dict(row)

if "fields" in row:

print("FIELDS_PRESENT")

sys.exit(4)

if len(row.get("content") or "") > 240:

print("PREVIEW_TOO_LONG")

sys.exit(5)

print("OK")

sys.exit(0)

"""Tb4b4b4,

Tb4b4b4)

Te6edf3result Tff7b72= Te6edf3subprocessTff7b72.Td2a8ffrunTb4b4b4(

Tb4b4b4[Te6edf3sysTff7b72.Td2a8ffexecutableTb4b4b4, Ta5d6ff"Ta5d6ff-cTa5d6ff"Tb4b4b4, Te6edf3scriptTb4b4b4]Tb4b4b4,

Te6edf3cwdTff7b72=Tffa657strTb4b4b4(Te6edf3PathTb4b4b4(Tff7b72__file__Tb4b4b4)Tff7b72.Td2a8ffresolveTb4b4b4(Tb4b4b4)Tff7b72.Td2a8ffparentsTb4b4b4[T79c0ff2Tb4b4b4]Tb4b4b4)Tb4b4b4,

Te6edf3capture_outputTff7b72=Tff7b72TrueTb4b4b4,

Te6edf3textTff7b72=Tff7b72TrueTb4b4b4,

Te6edf3timeoutTff7b72=T79c0ff60Tb4b4b4,

Te6edf3checkTff7b72=Tff7b72FalseTb4b4b4,

Tb4b4b4)

Tff7b72if Ta5d6ff"Ta5d6ffLANDLOCK_NOT_APPLIEDTa5d6ff" Tff7b72in Te6edf3resultTff7b72.Td2a8ffstdoutTb4b4b4:

Te6edf3pytestTff7b72.Td2a8ffskipTb4b4b4(Ta5d6ff"Ta5d6ffLandlock could not be applied in this environmentTa5d6ff"Tb4b4b4)

Tff7b72assert Te6edf3resultTff7b72.Td2a8ffreturncode Tff7b72== T79c0ff0Tb4b4b4, Tb4b4b4(

Ta5d6fffTa5d6ff"Ta5d6ffstdout=Tffd700{Te6edf3resultTff7b72.Td2a8ffstdoutTffd700!r}Ta5d6ff stderr=Tffd700{Te6edf3resultTff7b72.Td2a8ffstderrTffd700!r}Ta5d6ff code=Tffd700{Te6edf3resultTff7b72.Td2a8ffreturncodeTffd700}Ta5d6ff"

Tb4b4b4)

Tff7b72assert Ta5d6ff"Ta5d6ffOKTa5d6ff" Tff7b72in Te6edf3resultTff7b72.Td2a8ffstdout